package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.UserRegisterBean;
import com.atguigu.util.ClickHouseUtil_hjy;
import com.atguigu.util.DateFormatUtil_hjy;
import com.atguigu.util.KafkaUtil_wm;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

public class Dws_UserRegisterWindow {
    public static void main(String[] args) throws Exception {
        // TODO: 2023/3/25 用户注册需求   注册即新增
        // TODO: 2023/3/25 获取 flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism( 1 );
        // TODO: 2023/3/25 读取 kafka 主题  dwd_user_register 的数据
        String topic = "dwd_user_register";
        String groupId = "Dws_UserRegisterWindow";
        DataStreamSource<String> kafkaDS = env.addSource( KafkaUtil_wm.getFlinkKafkaConsumer( topic, groupId ) );
        // TODO: 2023/3/27 过滤并转为 javabaen 对象
        SingleOutputStreamOperator<UserRegisterBean> javaBeanDS = kafkaDS.map( line -> {
            JSONObject jsonObject = JSONObject.parseObject( line );
            return new UserRegisterBean( "", "",
                    1L,
                    DateFormatUtil_hjy.toTs( jsonObject.getString( "create_time" ), true )
            );
        } );
        // TODO: 2023/3/27 提取注册时间为watermark
        SingleOutputStreamOperator<UserRegisterBean> userRefisterWithWMDS = javaBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<UserRegisterBean>forBoundedOutOfOrderness( Duration.ofSeconds( 10 ) )
                        .withTimestampAssigner( new SerializableTimestampAssigner<UserRegisterBean>() {
                            @Override
                            public long extractTimestamp(UserRegisterBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        } )
        );

        // TODO: 2023/3/27 开窗聚合
        SingleOutputStreamOperator<UserRegisterBean> resultDS = userRefisterWithWMDS
                .windowAll( TumblingProcessingTimeWindows.of( Time.seconds( 3 ) ) )
                .reduce( new ReduceFunction<UserRegisterBean>() {
                             @Override
                             public UserRegisterBean reduce(UserRegisterBean value1, UserRegisterBean value2) throws Exception {
                                 value1.setRegisterCt( value1.getRegisterCt() + value2.getRegisterCt() );
                                 return value1;
                             }
                         },
                        new AllWindowFunction<UserRegisterBean, UserRegisterBean, TimeWindow>() {
                            @Override
                            public void apply(TimeWindow window, Iterable<UserRegisterBean> values, Collector<UserRegisterBean> out) throws Exception {
                                UserRegisterBean userRegisterBean = values.iterator().next();
                                userRegisterBean.setTs( System.currentTimeMillis() );
                                userRegisterBean.setStt( DateFormatUtil_hjy.toYmdHms( window.getStart() ) );
                                userRegisterBean.setEdt( DateFormatUtil_hjy.toYmdHms( window.getEnd() ) );

                                out.collect( userRegisterBean );
                            }
                        } );
        // TODO: 2023/3/27 将数据写出clickhouse
        resultDS.print( "用户注册窗口数据即将写出到clickhouse表：dws_user_user_register_window >>>>>>>>>>>>>>>>>>>>>" );
        resultDS.addSink( ClickHouseUtil_hjy.getClickHouseSink( "insert into dws_user_user_register_window values(?,?,?,?)" ) );
        // TODO: 2023/3/27 启动任务
        env.execute();
    }
}
